package com.lzwk.app.dwm;

import com.lzwk.app.function.GetJsonString;
import com.lzwk.app.ods.OdsOcpxPayOrder;
import com.lzwk.base.BaseSql;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @title: Dwm_ocpx_skin_trans
 * @Author Yeaser
 * @Date: 2022/1/5 17:29
 * @Version 1.0
 * Description: TODO
 */
public class DwmOcpxSkinTrans extends BaseSql {
    public static void main(String[] args) {
        new OdsOcpxPayOrder().init(4,"ck","ocpx_skin");
    }


    @Override
    protected void run(StreamTableEnvironment tenv) {

        tenv.getConfig().getConfiguration().setString("pipeline.name","ocpx_skin");
        //1.0 数据源
        tenv.executeSql("CREATE TABLE source_skinUV (" +
                "   ts STRING," +
                "   url_params STRING, " +
                "   tor_id STRING, " +
                "   spm STRING, " +
                "   evt STRING, " +
                "   process_time AS PROCTIME()" +
                " )WITH (" +
                "   'connector'='kafka', " +
                "   'topic'='earth_log_clean', " +
                "   'properties.bootstrap.servers'='10.1.128.74:9092', " +
                "   'properties.group.id'='ods_ocpx_skinUV_new', " +
                "   'scan.startup.mode'='earliest-offset', " +
                "   'format'='json'" +
                ")");

        tenv.createTemporaryFunction("GetJsonString", GetJsonString.class);

        Table source_ocpx_skinUV=tenv.sqlQuery(" select  " +
                " cast(to_date(ts) as string) dt, " +
                " GetJsonString(url_params,'source_id') as source_hash_id, " +
                " tor_id, " +
                " SPLIT_INDEX(spm, '.', 0) AS spm, " +
                " evt, " +
                " process_time " +
                " from " +
                " source_skinUV " +
                " where SPLIT_INDEX(spm, '.', 0) = 'oc' " +
                " and evt in ('ocpx_page_v','oc_page_v','oc_buy_ck_evt','oc_bp_se_v','om_bp_se_ev','om_bp_se_er_ev','oc_play_v','om_bp_ev','om_bp_er_ev','om_page_v') "
        );

        tenv.createTemporaryView("source_ocpx_skinUV",source_ocpx_skinUV);

        //2.1取出source信息 promote_promotesource
        tenv.executeSql(" CREATE TABLE mysql_promotesource ( " +
                "  id INT, " +
                "  skin_id INT, " +
                "  hash_id string, " +
                "  product_id INT, " +
                "  wxapp_id INT, " +
                "  name string, " +
                "  primary key (id) not enforced " +
                ") with ( " +
                "  'connector' = 'jdbc', " +
                "  'driver' = 'com.mysql.cj.jdbc.Driver', " +
                "  'url' = 'jdbc:mysql://10.1.192.67:3306/ocpx_elf?characterEncoding=utf8&serverTimezone=Asia/Shanghai&autoReconnect=true', " +
                "  'table-name' = 'promote_promotesource', " +
                "  'username'='ocpx_galaxy', " +
                "  'password'='FLsdf123', " +
                "  'lookup.cache.max-rows'='20', " +
                "  'lookup.cache.ttl'='600000' " +
                ")");


        //2.2 取出tag信息
        tenv.executeSql("CREATE TABLE mysql_tag_info (" +
                "  id INT, " +
                "  tag_name string, " +
                "  primary key (id) not enforced " +
                ") with ( " +
                "  'connector'='jdbc', " +
                "  'url'='jdbc:mysql://10.1.192.67:3306/ocpx_elf?characterEncoding=utf8&serverTimezone=Asia/Shanghai&autoReconnect=true', " +
                "  'table-name'='tag_info', " +
                "  'username'='ocpx_galaxy', " +
                "  'password'='FLsdf123', " +
                "  'lookup.cache.max-rows'='5', " +
                "  'lookup.cache.ttl'='600000' " +
                ")");

        //2.3 取出product信息
        tenv.executeSql("CREATE TABLE mysql_product (" +
                "  id INT, " +
                "  name string, " +
                "  tag_id INT, " +
                "  primary key (id) not enforced " +
                ") with ( " +
                "  'connector'='jdbc', " +
                "  'url'='jdbc:mysql://10.1.192.67:3306/ocpx_elf?characterEncoding=utf8&serverTimezone=Asia/Shanghai&autoReconnect=true', " +
                "  'table-name'='product_product', " +
                "  'username'='ocpx_galaxy', " +
                "  'password'='FLsdf123', " +
                "  'lookup.cache.max-rows'='10', " +
                "  'lookup.cache.ttl'='600000' " +
                ")");

        //2.4 取出微信app信息
        tenv.executeSql("CREATE TABLE mysql_wxapp (" +
                "  id INT, " +
                "  nickname string, " +
                "  primary key (id) not enforced " +
                ") with ( " +
                "  'connector'='jdbc', " +
                "  'url'='jdbc:mysql://10.1.192.67:3306/ocpx_elf?characterEncoding=utf8&serverTimezone=Asia/Shanghai&autoReconnect=true', " +
                "  'table-name'='wxapp_wxapp', " +
                "  'username'='ocpx_galaxy', " +
                "  'password'='FLsdf123', " +
                "  'lookup.cache.max-rows'='20', " +
                "  'lookup.cache.ttl'='600000' " +
                ")");

        // 2.5取出skin信息
        tenv.executeSql("CREATE TABLE mysql_productskin (" +
                "  id INT, " +
                "  name string, " +
                "  primary key (id) not enforced " +
                ") with ( " +
                "  'connector'='jdbc', " +
                "  'url'='jdbc:mysql://10.1.192.67:3306/ocpx_elf?characterEncoding=utf8&serverTimezone=Asia/Shanghai&autoReconnect=true', " +
                "  'table-name'='product_productskin', " +
                "  'username'='ocpx_galaxy', " +
                "  'password'='FLsdf123', " +
                "  'lookup.cache.max-rows'='20', " +
                "  'lookup.cache.ttl'='600000' " +
                ")");

        //2.6取出org信息
        tenv.executeSql("CREATE TABLE mysql_recordorgref (" +
                "  id int, " +
                "  object_type string, " +
                "  object_id int, " +
                "  org_id int, " +
                "  primary key (id) not enforced " +
                ") with ( " +
                "  'connector'='jdbc', " +
                "  'url'='jdbc:mysql://10.1.192.67:3306/ocpx_elf?characterEncoding=utf8&serverTimezone=Asia/Shanghai&autoReconnect=true', " +
                "  'table-name'='public_admin_recordorgref', " +
                "  'username'='ocpx_galaxy', " +
                "  'password'='FLsdf123', " +
                "  'lookup.cache.max-rows'='50', " +
                "  'lookup.cache.ttl'='600000' " +
                ")");

        //2.7 cdc读取业务库订单表，取出订单总数量和支付成功数量
        tenv.executeSql("create table mysql_pay_order(" +
                "   `id` INT, " +
                "   `source_id` INT, " +
                "   `pay_time`  TIMESTAMP(0), " +
                "   `create_time`  TIMESTAMP(0), " +
                "   `paid` INT, " +
                "   PRIMARY KEY (`id`) NOT ENFORCED" +
                ")with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='10.1.192.67', " +
                "   'port'='3306', " +
                "   'username'='ocpx_galaxy'," +
                "   'password'='FLsdf123'," +
                "   'database-name'='ocpx_elf'," +
                "   'table-name'='pay_order'," +
                "   'server-time-zone'='Asia/Shanghai'," +
                "   'debezium.snapshot.locking.mode' = 'none'" +
                ")");


        Table dwdPayOrder = tenv.sqlQuery(" select " +
                "   `source_id` , " +
                "   cast(to_date(cast(create_time as STRING)) as STRING) as dt ," +
                "   count(*) as total_order, " +
                "   sum(paid) as total_success_buy " +
                "   from  mysql_pay_order " +
                "   group by " +
                "   `source_id`, " +
                "   cast(to_date(cast(create_time as STRING)) as STRING) " );

        tenv.createTemporaryView("dwdPayOrder",dwdPayOrder);


        //3.0 UV指标计算
        Table ocpxSkinUVWithoutDim = tenv.sqlQuery(" select " +
                " dt, " +
                " source_hash_id, " +
                " count(distinct tor_id) filter (where evt in('ocpx_page_v')) as visit_page, " +
                " count(distinct tor_id) filter (where evt in('oc_page_v','om_page_v')) as success_visit_page, " +
                " count(distinct tor_id) filter (where evt in('oc_buy_ck_evt')) as bottom_button_click, " +
                " count(distinct tor_id) filter (where evt in('oc_bp_se_v')) as send_verify_code, " +
                " count(distinct tor_id) filter (where evt in('om_bp_se_ev')) as success_verify_code, " +
                " count(distinct tor_id) filter (where evt in('om_bp_se_er_ev')) as error_verify_code, " +
                " count(distinct tor_id) filter (where evt in('oc_play_v')) as audition_button_click, " +
                " count(distinct tor_id) filter (where evt in('om_bp_er_ev')) as total_error_buy, " +
                " max(process_time) as process_time " +
                " from source_ocpx_skinUV " +
                " group by " +
                " dt, " +
                " source_hash_id");



        tenv.createTemporaryView("ocpxSkinUVWithoutDim",ocpxSkinUVWithoutDim);

        //4.0 维度join
        Table skinWithDim = tenv.sqlQuery(" select " +
                " `t1`.`dt`, " +
                " `t7`.`org_id` as `org_id`, " +
                " `t2`.`id` as `source_id`, " +
                " `t3`.`tag_id` as `tag_id`, " +
                " `t2`.`wxapp_id` as `wxapp_id`, " +
                " `t2`.`product_id` as `product_id`, " +
                " `t2`.`skin_id` as `skin_id`, " +
                " regexp_replace(`t2`.`name`,'\\n|\\t','') as `source_name`, " +
                " regexp_replace(`t4`.`tag_name`,'\\n|\\t','') as `tag_name`, " +
                " regexp_replace(`t5`.`nickname`,'\\n|\\t','') as `wxapp_name`, " +
                " regexp_replace(`t3`.`name`,'\\n|\\t','') as `product_name`, " +
                " regexp_replace(`t6`.`name`,'\\n|\\t','') as `skin_name`, " +
                " `visit_page`, " +
                " `success_visit_page`, " +
                " `bottom_button_click`, " +
                " `send_verify_code`, " +
                " `success_verify_code`, " +
                " `error_verify_code`, " +
                " `audition_button_click`, " +
                " `total_error_buy` " +
                " from ocpxSkinUVWithoutDim t1 " +
                " left join mysql_promotesource for SYSTEM_TIME as of `t1`.`process_time` as t2 " +
                " on t1.source_hash_id=t2.hash_id " +
                " left join mysql_product for SYSTEM_TIME as of `t1`.`process_time` as t3 " +
                " on t2.product_id=t3.id " +
                " left join mysql_tag_info for SYSTEM_TIME as of `t1`.`process_time` as t4 " +
                " on t3.tag_id=t4.id " +
                " left join mysql_wxapp for SYSTEM_TIME as of `t1`.`process_time` as t5 " +
                " on t2.wxapp_id=t5.id " +
                " left join mysql_productskin for SYSTEM_TIME as of `t1`.`process_time` as t6 " +
                " on t2.skin_id=t6.id " +
                " left join mysql_recordorgref for SYSTEM_TIME as of `t1`.`process_time` as t7 " +
                " on t2.product_id=t7.object_id " +
                " and t7.object_type='product' "
        );

        tenv.createTemporaryView("skinWithDim",skinWithDim);

        //5.0 创建snik表
        tenv.executeSql("create table ads_ocpx_buried_trans(" +
                "   `dt` string, " +
                "   `org_id` INT, " +
                "   `source_id` INT, " +
                "   `tag_id` INT, " +
                "   `wxapp_id` INT, " +
                "   `product_id` INT, " +
                "   `skin_id` INT, " +
                "   `source_name` string, " +
                "   `tag_name` string, " +
                "   `wxapp_name` string, " +
                "   `product_name` string, " +
                "   `skin_name` string, " +
                "   `visit_page` BIGINT," +
                "   `success_visit_page` BIGINT," +
                "   `bottom_button_click` BIGINT," +
                "   `bottom_button_rate` decimal(16, 4)," +
                "   `send_verify_code` BIGINT, " +
                "   `success_verify_code` BIGINT, " +
                "   `error_verify_code` BIGINT, " +
                "   `success_verify_rate` decimal(16, 4), " +
                "   `audition_button_click` BIGINT, " +
                "   `total_success_buy` BIGINT, " +
                "   `total_order` BIGINT, " +
                "   `total_error_buy` BIGINT, " +
                "   `count_rate` decimal(16, 4), " +
                "   PRIMARY KEY (`dt`, `org_id`,`source_id`,`tag_id`,`wxapp_id`,`product_id`,`skin_id`) NOT ENFORCED" +
                ")with(" +
                "   'connector'='doris', " +
                "   'fenodes'='10.2.16.44:8030', " +
                "   'table.identifier'='warehouse.ads_ocpx_buried_trans', " +
                "   'username'='data_group'," +
                "   'password'='GTCStpJKpELuKFSs'," +
                "   'sink.batch.size'='80000', " +
                "   'sink.batch.interval'='3s' " +
                ")");

        // 6.0 插入数据
        tenv.executeSql(" insert into ads_ocpx_buried_trans " +
                " select " +
                " t1.dt as dt, " +
                " ifnull(org_id,1) as org_id, " +
                " ifnull(t1.source_id ,-1) as source_id, " +
                " ifnull(tag_id,-1) as tag_id, " +
                " ifnull(wxapp_id,-1) as wxapp_id, " +
                " ifnull(product_id,-1) as product_id, " +
                " ifnull(skin_id,-1) as skin_id, " +
                " ifnull(source_name,'') as source_name, " +
                " ifnull(tag_name,'') as tag_name, " +
                " ifnull(wxapp_name,'') as wxapp_name, " +
                " ifnull(product_name,'') as product_name, " +
                " ifnull(skin_name,'') as product_name, " +
                " visit_page, " +
                " success_visit_page, " +
                " bottom_button_click, " +
                " if(success_visit_page = 0,0,cast(bottom_button_click as double)/success_visit_page) as bottom_button_rate, " +
                " send_verify_code, " +
                " success_verify_code, " +
                " error_verify_code, " +
                " if(send_verify_code = 0,0,cast(success_verify_code as double)/send_verify_code) as success_verify_rate, " +
                " audition_button_click, " +
                " ifnull(t2.total_success_buy,0) as  total_success_buy ," +
                " ifnull(t2.total_order,0)  as total_order, " +
                " total_error_buy, " +
                " ifnull(if(total_order = 0,0,cast(total_success_buy as double)/total_order),0) as count_rate " +
                " from skinWithDim t1 " +
                " left join " +
                " dwdPayOrder t2" +
                " on t1.dt=t2.dt " +
                " and t1.source_id=t2.source_id "
        );
    }
}

